package cn.gwm.flink.streaming.function.process;

import cn.gwm.flink.streaming.beans.termalRunaway.BmsTrdStrategys;
import cn.gwm.flink.streaming.beans.termalRunaway.WaringTermalWithStragegyAndFlagEv;
import cn.gwm.flink.streaming.constant.TermalProcessResult;
import cn.gwm.flink.streaming.constant.TermalRunawayTypeEnum;
import cn.gwm.flink.streaming.constant.VehiclePowerType;
import cn.gwm.flink.streaming.strategy.termal.TermalStrategyFactory;
import cn.gwm.flink.streaming.task.ThermalWarningEvDwmTaskTest;
import cn.gwm.utils.StreamUtil;
import cn.hutool.core.lang.Dict;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.List;
import java.util.Objects;

/**
 * 热失控数据处理函数
 *
 * @author GW00283474
 */
public class TermalWarningRiseEvProcessFunction extends ProcessWindowFunction<Dict,Dict,String, GlobalWindow> {

    private static final Logger logger = LoggerFactory.getLogger(TermalWarningRiseEvProcessFunction.class);

    private ListState<WaringTermalWithStragegyAndFlagEv> listState;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        StateTtlConfig stateTtlConfig = StateTtlConfig
                // 状态有效时间    1天过期
                .newBuilder(Time.days(1))
                // 设置状态的更新类型
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                // 已过期还未被清理掉的状态数据不返回给用户
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                // 过期对象的清理策略 全量清理
                .cleanupFullSnapshot()
                .build();
        ListStateDescriptor termalWarningRiseDesc = new ListStateDescriptor("termalWarningRiseDescEv", WaringTermalWithStragegyAndFlagEv.class);
        termalWarningRiseDesc.enableTimeToLive(stateTtlConfig);
        listState = getRuntimeContext().getListState(termalWarningRiseDesc);

    }

    @Override
    public void process(String s, ProcessWindowFunction<Dict, Dict, String, GlobalWindow>.Context ctx, Iterable<Dict> elements, Collector<Dict> out) throws Exception {
        try {
            List<Dict> dictList = StreamUtil.convertIterable(elements);
            if(dictList.size()>1){
                logger.info("EV进入升温、电压变化校验中，数据为：{}", JSONUtil.toJsonPrettyStr(dictList));
                Iterator<Dict> iterator = elements.iterator();
                Dict lastDict=iterator.next();
                Dict currentDict=iterator.next();
                WaringTermalWithStragegyAndFlagEv last = (WaringTermalWithStragegyAndFlagEv) lastDict.get(TermalProcessResult.CHECK_RECORD);
                WaringTermalWithStragegyAndFlagEv current = (WaringTermalWithStragegyAndFlagEv)currentDict.get(TermalProcessResult.CHECK_RECORD);
                listState.add(last);listState.add(current);
                //温升校验策略
                BmsTrdStrategys temperatureRiseStrategys = current.getBmsTrdStrategies().stream().filter(el -> "1".equals(el.getState()) && TermalRunawayTypeEnum.TEMPERATURERISE.getKey().equals(el.getItemType())).findFirst().orElseGet(() -> {
                    return null;
                });
                logger.info("EV温升校验完成");
                //电压变化校验策略
                BmsTrdStrategys voltageVariationStrategys = current.getBmsTrdStrategies().stream().filter(el -> "1".equals(el.getState()) && TermalRunawayTypeEnum.VOLTAGEVARIATION.getKey().equals(el.getItemType())).findFirst().orElseGet(() -> {
                    return null;
                });
                logger.info("EV电压变化校验完成");
                if(Objects.nonNull(temperatureRiseStrategys)){
                    Dict tempRiseCheck = TermalStrategyFactory.getTermalStrategy(TermalRunawayTypeEnum.TEMPERATURERISE.getKey(), false, VehiclePowerType.EV).check(listState, 0);
                    if(Objects.nonNull(tempRiseCheck) && !tempRiseCheck.isEmpty()){
                        currentDict.set(TermalProcessResult.DICT_TEMPRISE,tempRiseCheck);
                    }
                }

                if(Objects.nonNull(voltageVariationStrategys)){
                    Dict voltChanageCheck = TermalStrategyFactory.getTermalStrategy(TermalRunawayTypeEnum.VOLTAGEVARIATION.getKey(), false,VehiclePowerType.EV).check(listState, 0);
                    if(Objects.nonNull(voltChanageCheck) && !voltChanageCheck.isEmpty()){
                        currentDict.set(TermalProcessResult.DICT_VOLTCHANAGE,voltChanageCheck);
                    }
                }
                listState.clear();
                logger.info("EV升温、电压变化全部校验完成");
                out.collect(currentDict);
            }
        }catch (Exception e){
            e.printStackTrace();
            logger.error("EV温升、电压变化策略校验异常message:{},case:{},stackTrace:{}",e.getMessage(),e.getCause(),e.getStackTrace());
        }
    }
}
